package com.shujia.hbase.javaapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;


public class Demo1 {
    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {
        /**
         * 通过java代码操作hbase
         *
         * 1、建立连接
         * 2、对数据和 表进行操作
         * 3、关闭连接
         *
         */

        //创建hadoop配置文件对象
        Configuration conf = new Configuration();

        //指定zk连接地址
        conf.set("hbase.zookeeper.quorum", "node2:2181,node3:2181.node4:2181");

        //创建连接

        /**
         * 对表进行操作连接maste
         *
         * 对数据进行操作连接regionserver
         *
         */

        try {
            //增删改数据使用
            connection = HConnectionManager.createConnection(conf);

            //对表操作时使用
            hBaseAdmin = new HBaseAdmin(conf);
            System.out.println("连接创建成功");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void createTable() {

        /**
         * 创建表
         *
         * create 'table','info'
         * create 'user',{NAME => 'info',TTL => 5}
         *
         */
        //表描述对象
        HTableDescriptor hTableDescriptor = new HTableDescriptor("student");
        //列簇描述对象
        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor("info");
        //指定版本数量
        hColumnDescriptor.setMaxVersions(10);
        //指定数据过期时间
        //hColumnDescriptor.setTimeToLive(5);

        //增加列簇
        hTableDescriptor.addFamily(hColumnDescriptor);

        try {
            //判断是否存在
            boolean tableExists = hBaseAdmin.tableExists("student");

            if (!tableExists) {
                //创建表   由master
                hBaseAdmin.createTable(hTableDescriptor);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Test
    public void deleteTable() {
        try {
            //禁用表
            hBaseAdmin.disableTable("student");
            //删除表
            hBaseAdmin.deleteTable("student");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void listTable() {
        try {
            //获取所有表
            HTableDescriptor[] hTableDescriptors = hBaseAdmin.listTables();
            for (HTableDescriptor tableDescriptor : hTableDescriptors) {
                byte[] name = tableDescriptor.getName();
                //将字节数组转换成字符串
                String tn = new String(name, "utf-8");
                System.out.println(tn);

            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Test
    public void insert() {

        try {
            /**
             * 获取表实例对象
             *
             * put 'table','row1','info:name','value'
             *
             */
            HTableInterface student = connection.getTable("student");

            //构建put对象
            Put put = new Put("002".getBytes());
            put.add("i".getBytes(), "name".getBytes(), "zhangsan".getBytes());

            student.put(put);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Test
    public void insertAll() throws Exception {
        /**
         * 批量插入数据
         *
         */

        ArrayList<Put> puts = new ArrayList<>();


        BufferedReader bf = new BufferedReader(new FileReader("C:\\bigdara\\java\\data\\students.txt"));
        String line;
        while ((line = bf.readLine()) != null) {

            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];

            //构建put对象  以学号作为rowkey
            Put put = new Put(id.getBytes());
            //增加列
            put.add("info".getBytes(), "name".getBytes(), name.getBytes());
            put.add("info".getBytes(), "age".getBytes(), age.getBytes());
            put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());


            puts.add(put);
        }

        //获取表的对象
        HTableInterface student = connection.getTable("student");
        //批量插入数据
        student.put(puts);


    }

    @Test
    public void query() throws Exception {
        /**
         * 查询数据
         *
         * get 'table',"row1"
         */

        //获取表实例
        HTableInterface student = connection.getTable("student");

        //构建get对象  指定rowkey
        Get get = new Get("1500100999".getBytes());

        //指定需要查询的列簇
        get.addFamily("info".getBytes());

        //指定查询某一列的数据
        // get.addColumn("info".getBytes(), "name".getBytes());

        //查询数据   result结果对象
        Result result = student.get(get);

        //从结果中获取一个单元格的数据
        byte[] value = result.getValue("info".getBytes(), "name".getBytes());

        // new String(value,"utf-8")

        //将字节数组转换成字符串
        String name = Bytes.toString(value);
        System.out.println(name);

        String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));

        System.out.println(age);


        //获取所有列
        List<Cell> cells = result.listCells();

        for (Cell cell : cells) {
            //取出value
            byte[] v = CellUtil.cloneValue(cell);
            String s = Bytes.toString(v);
            System.out.println(s);

            byte[] bytes = CellUtil.cloneFamily(cell);
            String f = Bytes.toString(bytes);
            System.out.println(f);

            System.out.println("==========");

        }
    }

    @Test
    public void scan() throws Exception {
        //获取表实例
        HTableInterface student = connection.getTable("student");

        //构建scan对象
        Scan scan = new Scan();

        //指定列簇扫描
        scan.addFamily("info".getBytes());

        //返回一个迭代器
        ResultScanner results = student.getScanner(scan);

        /**
         * Result   代表一行数据   一个rowkey的数据
         *
         */

        for (Result result : results) {
            //取出rowkey
            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }

    /**
     * 取出文科的学生
     * <p>
     * 使用列值过滤器  使用正则比较器获取前缀比较器
     */

    @Test
    public void filterClazz() throws Exception {

        //获取表实例
        HTableInterface student = connection.getTable("student");

        //构建scan对象
        Scan scan = new Scan();


        /**
         * 构建过滤器对象
         *SingleColumnValueFilter  列值过滤器   会进行全表扫描
         *
         */

        //正则比较器
        //RegexStringComparator regexStringComparator = new RegexStringComparator("文科..");


        //前缀比较器
        BinaryPrefixComparator prefixComparator = new BinaryPrefixComparator("文科".getBytes());


        SingleColumnValueFilter filter = new SingleColumnValueFilter(
                "info".getBytes(),//列簇
                "clazz".getBytes(),//列名
                CompareFilter.CompareOp.EQUAL,//比较规整
                prefixComparator//比较器
        );

        //增加过滤器
        scan.setFilter(filter);


        //返回一个迭代器
        ResultScanner results = student.getScanner(scan);

        /**
         * Result   代表一行数据   一个rowkey的数据
         *
         */

        for (Result result : results) {
            //取出rowkey
            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }

    /**
     * 列簇过滤器
     */
    @Test
    public void filter() throws Exception {

        //获取表实例
        HTableInterface student = connection.getTable("student");

        //构建scan对象
        Scan scan = new Scan();


        /**
         * 增加列簇过滤器
         *
         */
        //前缀比较器
        // BinaryPrefixComparator prefixComparator = new BinaryPrefixComparator("in".getBytes());

        // FamilyFilter filter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, prefixComparator);

        //前缀比较器
        BinaryPrefixComparator prefixComparator = new BinaryPrefixComparator("n".getBytes());
        //列名过滤器
        QualifierFilter filter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, prefixComparator);

        scan.setFilter(filter);


        //返回一个迭代器
        ResultScanner results = student.getScanner(scan);

        /**
         * Result   代表一行数据   一个rowkey的数据
         *
         */

        for (Result result : results) {
            //取出rowkey
            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }


    /**
     * 取出rowkey里面以1结尾的所有数据
     */
    @Test
    public void rowFilter() throws Exception {
        //获取表实例
        HTableInterface student = connection.getTable("student");

        //构建scan对象
        Scan scan = new Scan();

        /**
         * 增加行键过滤器
         * 取出rowkey里面以1结尾的所有数据
         */

        //正则比较器
        RegexStringComparator regexStringComparator = new RegexStringComparator(".{9}1");

        //行键过滤器
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, regexStringComparator);

        scan.setFilter(rowFilter);

        //返回一个迭代器
        ResultScanner results = student.getScanner(scan);

        /**
         * Result   代表一行数据   一个rowkey的数据
         *
         */

        for (Result result : results) {
            //取出rowkey
            String id = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }

    @After
    public void close() {

        try {
            hBaseAdmin.close();
            connection.close();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
